package org.cancer.service

import org.apache.spark.streaming.dstream.DStream
import org.cancer.dao.HighRiskDao
import org.cancer.util.SparkUtil
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.DataFrame
import org.cancer.bean.HighRiskIdentificationData

class HighRiskIdentificationService_Sylvia {
  private val dao = new HighRiskDao()
  private val spark = SparkUtil.takeSpark()

  def dataAnalysis(data: DStream[HighRiskIdentificationData]): Unit = {
    println("开始分析高危患者数据...")

    // 输出每条原始数据，便于核查字段内容
    /*data.foreachRDD { rdd =>
      rdd.collect().foreach { patient =>
        println(s"原始数据: ts=${patient.ts}, installment=${patient.Installment}, transfer=${patient.transfer}, complications=${patient.Complications}")
      }
    }*/

    // 筛选出满足条件的高危患者
    val highRiskPatients = data.filter { patient =>
      val installment = patient.Installment.trim
      val transfer = patient.transfer.trim
      val complications = patient.Complications.trim

      val isAdvancedStage = installment == "III" || installment == "IV"
      val hasTransfer = transfer == "是"
      val hasComplications = complications != "无" && complications.nonEmpty

      // 调试输出每条数据的判断过程
      println(s"判定: isAdvancedStage=$isAdvancedStage, hasTransfer=$hasTransfer, hasComplications=$hasComplications")

      isAdvancedStage && hasTransfer && hasComplications
    }

    // 处理每个批次的数据
    highRiskPatients.foreachRDD(rdd => {
      val total = rdd.count()
      if (total > 0) {
        rdd.collect().foreach(p => println(s"高危患者数据: $p"))
        import spark.implicits._
        val df: DataFrame = spark.createDataFrame(rdd)

        println(s"发现 $total 例高危患者，正在存入数据库...")
        df.show()

        // 存入数据库
        try {
          df.write
            .format("jdbc")
            .option("url", "jdbc:mysql://node1:3306/cancer_patients")
            .option("driver", "com.mysql.cj.jdbc.Driver")
            .option("user", "root")
            .option("password", "123456")
            .option("dbtable", "Patient_Data_Sylvia")
            .mode(SaveMode.Append)
            .save()
          println("数据成功写入数据库")
        } catch {
          case e: Exception =>
            println(s"写入数据库失败: ${e.getMessage}")
            e.printStackTrace()
        }
      } else {
        println("本批次无高危患者数据")
      }
    })
  }
}

